package cn.deschen.search.mysql.senders;

import cn.deschen.biz.constants.MsgConstants;
import cn.deschen.biz.domain.entity.MsgRecord;
import cn.deschen.biz.service.MsgRecordService;
import cn.deschen.search.mysql.BinlogRowData;
import cn.deschen.search.mysql.handlers.BinlogRowDataHandlerFactory;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Optional;

/**
 * @Author hanbin_chen
 * @Description 索引数据Kafka异步消息处理
 * @DateTime 2021/6/10
 * @Version V1.0.0
 */
@Slf4j
public class IndexKafkaSender implements IndexSender {

    @Value("${kafka.topic}")
    private String topic;

    private KafkaTemplate<String, String> kafkaTemplate;

    private MsgRecordService msgRecordService;

    public IndexKafkaSender(KafkaTemplate<String, String> kafkaTemplate, MsgRecordService msgRecordService) {
        this.kafkaTemplate = kafkaTemplate;
        this.msgRecordService = msgRecordService;
    }

    @Override
    public void sender(BinlogRowData rowData) {
        msgRecordService.sendToData(rowData.getMsgId(), JSONUtil.toJsonStr(rowData));
        log.debug("生产者 msgId={} rowData={}", rowData.getMsgId(), rowData);
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, rowData.getMsgId(), JSONUtil.toJsonStr(rowData));
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                msgRecordService.remarkSendSuccess(rowData.getMsgId());
                log.debug("生产者成功 msgId={} rowData={}", rowData.getMsgId(), rowData);
            }

            @Override
            public void onFailure(Throwable ex) {
                msgRecordService.remarkSendError(rowData.getMsgId(), ex.getMessage());
                log.debug("生产者失败 msgId={} rowData={} exception={}", rowData.getMsgId(), rowData, ex.getMessage());
                throw new RuntimeException("Error sending message", ex);
            }
        });
    }

    @KafkaListener(topicPartitions = {
            @TopicPartition(topic = "ad-search", partitions = {"0", "1", "2"})
    }, groupId = "ad-search-group", id = "consumer00")
    public void processMysqlRowData0(@Header(KafkaHeaders.OFFSET) Integer offset,
                                     @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                     @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String msgId,
                                     ConsumerRecord<?, ?> record,
                                     Acknowledgment ack){
        log.info("消费者{} msgId={} record={} offset={} partition={}", Thread.currentThread().getName(), msgId, record.value(), offset, partition);
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            BinlogRowData rowData = JSONUtil.toBean(message.toString(), BinlogRowData.class);
            try {
                BinlogRowDataHandlerFactory.getHandler(rowData.getDatabase(), rowData.getTableName()).handlerBinlogRowData(rowData);
                msgRecordService.remarkConsumerSuccess(msgId);
                log.info("消费者成功{} msgId={} status={} record={} offset={} partition={}",
                        Thread.currentThread().getName(), msgId, record.value(), offset, partition);
            } catch (Exception e) {
                String exception = e.getMessage();
                msgRecordService.remarkConsumerError(msgId, exception);
                log.info("消费者失败{} msgId={} status={} record={} offset={} partition={} exception={}",
                        Thread.currentThread().getName(), msgId, record.value(), offset, partition, exception);
            } finally {
                // 手动提交偏移量
                ack.acknowledge();
            }
        }

    }
}
